package com.bw.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Random;

public class BaseLogApp {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);


        String topic="topic_log";
        String groupId="base_log_consumer";

        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
//        source.print();
        OutputTag<String> dirtyStreamTag = new OutputTag<String>("dirtyStream") {};
        SingleOutputStreamOperator<String> cleanedStream = source.process(new ProcessFunction<String, String>() {

            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                try {
                    JSONObject jsonObj = JSON.parseObject(s);
                    JSONObject common = jsonObj.getJSONObject("common");
                    common.put("shop_id", new Random().nextInt(10 + 1));
                    int i = new Random().nextInt(2);
                    common.put("device_type", (i == 0 ? 0 : 1));
                    collector.collect(jsonObj.toJSONString());
                } catch (Exception e) {
                    context.output(dirtyStreamTag,s);
                    System.out.println(s);
                }
            }
        });
//        cleanedStream.print();

        DataStream<String> dirtyStream = cleanedStream.getSideOutput(dirtyStreamTag);
        String topic_dirty="dirty_data";
        dirtyStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(topic_dirty));
        dirtyStream.print("脏数据");
        SingleOutputStreamOperator<JSONObject> mappedStream = cleanedStream.map(JSON::parseObject);

        //新老访客
        /**
         * 1.我们的数据按照key分组
         * 2.给每个用户保存状态
         * 3.首日登录的用户 1
         * 看状态中是否有值 如果没有null 把今天的日期加进去
         * 如果有值判断值是否相同 如果相等变成1
         * 如果不相同 1变成0
         */


        //根据mid分组
        KeyedStream<JSONObject, String> keyedStream = mappedStream.keyBy(r -> r.getJSONObject("common").getString("mid"));//mid相同的给共享

        //新老访客状态标记修复

        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            //创建状态 给相同mid共享 容器里面只存日期  value  list map

            ValueState<String> firstViewDtState;

            @Override
            public void open(Configuration param) throws Exception {
                super.open(param);

                firstViewDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstViewDtState", String.class));
            }
            //zs is_new=1   当前时间=dt  状态值=firstViewDt

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                String isNew = jsonObject.getJSONObject("common").getString("is_new");
                // 1 0
                //状态时间
                String firstViewDt = firstViewDtState.value();
                //获取当前数据的时间
                Long ts = jsonObject.getLong("ts");
                //把时间戳转换成年月日
                String dt = DateFormatUtil.toDate(ts);
                if ("1".equals(isNew)) {
                    if (firstViewDt == null) {
                        firstViewDtState.update(dt);
                    } else {
                        if (!firstViewDt.equals(dt)) {
                            isNew = "0";
                            jsonObject.getJSONObject("common").put("is_new", isNew);
                        }
                    }
                } else {
                    if (firstViewDt == null) {
                        //将首次访问日期置为昨日
                        String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                        firstViewDtState.update(yesterday);
                    }
                }
                collector.collect(jsonObject);
            }
        });

        //todo 6.分流 page start err
        //定义启动、曝光、动作、错误侧输出流
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {};

        /*
        启动 common start ts err
        页面 common page display action ts err
        *
        * */
        //分流
        SingleOutputStreamOperator<String> separatedStream = fixedStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                //收集错误数据
                JSONObject error = jsonObject.getJSONObject("err");
                if (error != null) {
                    context.output(errorTag, jsonObject.toJSONString());
                }
                //剔除 "err" 字段
                jsonObject.remove("err");

                //收集启动数据
                JSONObject start = jsonObject.getJSONObject("start");
                if (start != null) {
                    context.output(startTag, jsonObject.toJSONString());
                } else {
                    //page display action
                    //获取"page" 字段
                    JSONObject page = jsonObject.getJSONObject("page");
                    //获取"common" 字段
                    JSONObject common = jsonObject.getJSONObject("common");
                    //获取"ts" 字段
                    Long ts = jsonObject.getLong("ts");


                    //收集曝光数据 [{},{},{}]
                    JSONArray displays = jsonObject.getJSONArray("displays");
                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            //创建空的json对象
                            JSONObject displayObj = new JSONObject();
                            displayObj.put("display", display);
                            displayObj.put("common", common);
                            displayObj.put("page", page);
                            displayObj.put("ts", ts);
                            context.output(displayTag, displayObj.toJSONString());
                        }
                    }
                    //收集动作数据
                    JSONArray actions = jsonObject.getJSONArray("actions");
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            JSONObject actionObj = new JSONObject();
                            actionObj.put("action", action);
                            actionObj.put("common", common);
                            actionObj.put("page", page);
                            actionObj.put("ts", ts);
                            context.output(actionTag, actionObj.toJSONString());
                        }
                    }
                    //收集页面数据
                    jsonObject.remove("displays");
                    jsonObject.remove("actions");


                    collector.collect(jsonObject.toJSONString());
                }
            }
        });
        //打印主流和各侧输出流查看分流效果
        separatedStream.print("page>>>");
        separatedStream.getSideOutput(startTag).print("start!!!");
        separatedStream.getSideOutput(displayTag).print("display@@@");
        separatedStream.getSideOutput(actionTag).print("action###");
        separatedStream.getSideOutput(errorTag).print("error$$$");

        //todo 7.将数据输出到kafka的不同主题
        //7.1 提取各侧输出流
        DataStream<String> startDS = separatedStream.getSideOutput(startTag);
        DataStream<String> displayDS = separatedStream.getSideOutput(displayTag);
        DataStream<String> actionDS = separatedStream.getSideOutput(actionTag);
        DataStream<String> errorDS = separatedStream.getSideOutput(errorTag);

        separatedStream.print("page>>>>>>>>>>>>>>>");
        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_page_log"));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_start_log"));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_display_log"));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_action_log"));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_error_log"));
        env.execute();
    }
}
